package com.shujia.flink.state;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo02WCCheckPoint {
    public static void main(String[] args) throws Exception {
        // 1、初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        // 每 10000ms 开始一次 checkpoint
        env.enableCheckpointing(10000);
//        // 高级选项：
//        // 设置模式为精确一次 (这是默认值)
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 确认 checkpoints 之间的时间会进行 500 ms
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        // 允许两个连续的 checkpoint 错误
//        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
//        // 同一时间只允许一个 checkpoint 进行
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //将状态保存到hdfs中
        env.getCheckpointConfig().setCheckpointStorage("hdfs://master:9000/flink/checkpoint");

        // 2、通过nc命令模拟一个流，使用env加载流中的数据
        DataStream<String> ds = env.socketTextStream("master", 8888);

        // 3、对每行数据进行切分 将每个单词变成一行
        // 需要指定返回值的类型
        /*
         * 可以通过lambda表达式对代码进行简化
         * 如果直接使用lambda表达式要注意指定类型
         */
        DataStream<String> wordsDS = ds.flatMap((line, collector) -> {
            String[] splits = line.split(",");
            // 遍历每一个单词
            for (String word : splits) {
                // 使用collector将结果数据发送给下游
                collector.collect(word);
            }
            // java中的lambda表达式有时无法自动推断类型，需要手动指定
        }, Types.STRING);

        // 4、将每个word变成word，1这种形式
        DataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));

        // 5、按照每个单词进行分组
        KeyedStream<Tuple2<String, Integer>, String> grpDS = kvDS.keyBy(kv -> kv.f0);

        // 6、统计单词数量
        DataStream<Tuple2<String, Integer>> wordCntDS = grpDS.sum(1);

        wordCntDS.print();

        // 启动任务
        env.execute();


    }
}
